package com.atguigu.app;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.atguigu.constants.GmallConstants;
import com.atguigu.utils.MyKafkaSender;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Random;

public class CanalClient {
    public static void main(String[] args) throws InvalidProtocolBufferException {
        //1.获取连接器
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("hadoop102", 11111), "example", "", "");

        while (true) {
            //2.获取连接
            connector.connect();

            //3.选择订阅的数据库
            connector.subscribe("gmall211227.*");

            //4.获取多个sql执行结果
            Message message = connector.get(100);

            //5.获取一个sql执行的结果
            List<CanalEntry.Entry> entries = message.getEntries();

            //6.判断mysql中是否有数据变化
            if (entries.size() <= 0) {
                System.out.println("没有数据休息一会。。。。");
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            } else {
                //7.获取每一个Entry
                for (CanalEntry.Entry entry : entries) {

                    //TODO 8.获取表名
                    String tableName = entry.getHeader().getTableName();

                    //9.根据entry类型获取数据
                    CanalEntry.EntryType entryType = entry.getEntryType();

                    if (entryType.equals(CanalEntry.EntryType.ROWDATA)) {
                        //10.获取序列化数据
                        ByteString storeValue = entry.getStoreValue();

                        //11.对数据做反序列化
                        CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(storeValue);
                        //TODO 12.获取事件类型
                        CanalEntry.EventType eventType = rowChange.getEventType();

                        //TODO 13.获取具体的数据
                        List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();

                        //根据不同的表名不同的事件类型获取对应的数据
                        handle(tableName, eventType, rowDatasList);
                    }
                }
            }

        }
    }

    private static void handle(String tableName, CanalEntry.EventType eventType, List<CanalEntry.RowData> rowDatasList) {
        //获取订单表新增的数据
        if ("order_info".equals(tableName) && CanalEntry.EventType.INSERT.equals(eventType)) {
            //遍历rowDatasList获取到每一行数据
            saveToKafka(rowDatasList, GmallConstants.KAFKA_TOPIC_ORDER);
        } else if ("order_detail".equals(tableName) && CanalEntry.EventType.INSERT.equals(eventType)) {
            //遍历rowDatasList获取到每一行数据
            saveToKafka(rowDatasList, GmallConstants.KAFKA_TOPIC_ORDER_DETAIL);
        } else if ("user_info".equals(tableName) && (CanalEntry.EventType.INSERT.equals(eventType)) || CanalEntry.EventType.UPDATE.equals(eventType)) {
            //遍历rowDatasList获取到每一行数据
            saveToKafka(rowDatasList, GmallConstants.KAFKA_TOPIC_USER);
        }
    }

    //选中相同的代码 按住ctrl+alt+m 生成对应的方法，然后全部替换
    private static void saveToKafka(List<CanalEntry.RowData> rowDatasList, String kafkaTopicOrder) {
        for (CanalEntry.RowData rowData : rowDatasList) {
            //获取到一行中每一列的数据
            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
            JSONObject jsonObject = new JSONObject();
            for (CanalEntry.Column column : afterColumnsList) {
                jsonObject.put(column.getName(), column.getValue());
            }
            System.out.println(jsonObject.toJSONString());
//            //将数据发送至kafka
//            //模拟网络震荡
//            try {
//                Thread.sleep(new Random().nextInt(5)*1000);
//            } catch (InterruptedException e) {
//                e.printStackTrace();
//            }
            MyKafkaSender.send(kafkaTopicOrder, jsonObject.toJSONString());
        }
    }
}
